/**提高消息可靠性的方法
 * 1.消息可靠性发送:事务机制和发送方确认 (publisher confirm) 机制
 * 2.消息可靠性接收：ACK机制
 * 3.持久化存储：交换器、队列、消息均可定义成持久化方式
 * 4.死信
 * 5.延迟消息队列
 */
package com.example.rabbitMQ;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

public class AdvancedDemo {
    private static final int count = 10;
    static Connection connection = null;
    static Channel channel = null;

    //生产者可靠性消息投递:生产者在发送消息过程中也可能出现错误或者网络延迟灯故障，导致消息未成功发送到交换机或者队列
    //下述方法只能保证 msg（消息） 到 Exchange（交换器）的可靠投递

    //方法1：事务机制（二选一），性能低,一条一条发送，一条一条确认
    //1.channel.txSelect 开启事务
    //2.channel.txCommit 提交事务
    //3.channel.txRollback 事务回滚
    private static final String QUEUE_NAME_TRANSACTION = "queue_transaction";
    //--------------------事务处理--------------------
    public static void transactionProducer() {
        int j;
        Channel channel = null;
        Connection connection = null;
        try {
            connection = BasicDemo.getConnection();
            channel = connection.createChannel();

            channel.queueDeclare(QUEUE_NAME_TRANSACTION, true, false, false, null);
            //开启事务
            channel.txSelect();
            for (int i = 0; i < count; i++) {
                try {
                    channel.basicPublish("", QUEUE_NAME_TRANSACTION, MessageProperties.PERSISTENT_TEXT_PLAIN, ("事务测试数据：" + (i + 1)).getBytes());
                    if (i == 4) j = 1 / 0;
                    //提交事务，提交之后会将内存中的消息写入队列，并释放内存
                    channel.txCommit();
                }catch (Exception e) {
                    //回滚事务，放弃当前事务中所有还未提交的消息，释放内存
                    channel.txRollback();
                    System.out.println("txRollback====");
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                channel.close();
                connection.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }

    }

    //方法2：发送方确认机制（二选一），性能高（比事务模式高一个数量级），当Channel设置成confirm模式时，发布的每一条消息都会获得一个唯一的deliveryTag
    //1、channel.confirmSelect
    //2、channel.waitForConfirms/waitForConfirmsOrDie/addConfirmListener
    private static final String QUEUE_NAME_CONFIRM = "queue_confirm";
    //--------------------发送方确认--------------------
    public static void confirmProducer() {
        int j;
        try {
            connection = BasicDemo.getConnection();
            channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME_CONFIRM, true, false, false, null);
            //启动发送者确认模式
            channel.confirmSelect();
//            channel.basicPublish("", QUEUE_NAME_CONFIRM, MessageProperties.PERSISTENT_TEXT_PLAIN, ("确认模式测试数据1").getBytes());
//            //普通Confirm模式，消息发送成功返回true；否则返回false；如果未在超时时间内确认该消息，将引发超时的异常
              //发送一条确认一条
//            if (channel.waitForConfirms(1000)) {
//                System.out.println("confirm消息发送成功");
//            } else {
//                System.out.println("confirm消息发送失败");
//                //重发消息
//            }

            //批量Confirm模式，只要有一个未确认就会IOException;发送一批，确认一批
            for (int i = 0; i < count; i++) {
                channel.basicPublish("", QUEUE_NAME_CONFIRM, null, ("确认模式测试数据2-----"+(i+1)).getBytes("UTF-8"));
            }
//            //批量确认，如果其中有一条消息没有发送成功或者服务器出现问题或超时都会报异常
//            channel.waitForConfirmsOrDie(3000);

            //异步Confirm模式，是否批量确认在于返回的multiple的参数，如果true表示批量执行了deliveryTag这个值以前的所有消息，如果为false的话表示单条确认
            //一直发送msg（消息），根据 RabbitMQ 发送过来的确认信息，一批一批进行确认
            channel.addConfirmListener(new ConfirmListener() {
                //deliveryTag:被确认消息的编号,从1开始自动递增，用于标记当前是第几个消息; multiple: 是否同时确认了多条消息
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("服务端 ACK Success,已发送成功消息--------"+deliveryTag+"-------"+multiple);
                }
                //multiple：true，小于当前编号的所有消息可能都没有发送成功；false：当前编号的消息没有发送成功
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.err.println("服务端 ACK Failed，发送失败消息--------"+deliveryTag+"-------"+multiple);
                }
            });
//            channel.close();
//            connection.close();

        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void txcfmConsumer() {
        try {
            connection = BasicDemo.getConnection();
            channel = connection.createChannel();
//            channel.queueDeclare(QUEUE_NAME_TRANSACTION, true, false, false, null);
            channel.queueDeclare(QUEUE_NAME_CONFIRM, true, false, false, null);
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("[生产者高可靠]received ===> " + new String(body));
                }
            };
//        channel.basicConsume(QUEUE_NAME_TRANSACTION, true, defaultConsumer);
          channel.basicConsume(QUEUE_NAME_CONFIRM, false, defaultConsumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    //消费者可靠性消息消费：消息确认机制（ACK）。收到消息时，先不回复确认，而是先处理业务逻辑，处理成功后回复确认，此时MQ会将该消息删除；业务逻辑处理失败则回复拒绝，然后将消息登记到异常表，由专门的异常处理程序处理。
    //channel.basicConsume第二个参数autoAck=true：自动确认（无需确认）；false：手动确认，通过手动确认来保证确实收到信息，且已被正确地处理完毕。未经确认的消息会依然保留在队列中，被手动拒绝的消息会被丢弃或返回队列
     private static final String QUEUE_NAME_ACK = "queue_ack";
    //--------------------ACK消息确认机制--------------------
    public static void ackProducer() {
        try {
            connection = BasicDemo.getConnection();
            channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME_ACK, false, false, false, null);
            for (int i = 0; i < count; i++) {
                channel.basicPublish("", QUEUE_NAME_ACK, null, ("ACK模式发送数据-----" + (i + 1)).getBytes());
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    static boolean isFirst=true;
    public static void ackConsumer() {
        try {
            connection = BasicDemo.getConnection();
            channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME_ACK, false, false, false, null);
            // 消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    try {
                        int i ;
                        String message = new String(body, "UTF-8");
                        if (message.contains("5") && isFirst) {
                            isFirst=false;
                            i = 1 / 0;  //模拟接收消息出错
                        }
                        System.out.println("[ACK手动确认消费者]received ===> " + message);
                        // 手动ACK，multiple=false，仅确认当前消息；true，确认所有小于等于当前id的消息
                        getChannel().basicAck(envelope.getDeliveryTag(), false);
                    } catch (Exception e1) {
                        e1.printStackTrace();
                        try {
                            //方式1：单个处理；参数2 标识是否重回队列 true 是重回  false 是不重回:丢弃消息，如果重回队列的话，异常没有解决，就会进入死循环
                            channel.basicReject(envelope.getDeliveryTag(), true);
                            //方式2：可以批量处理；参数3 标识是否重回队列 true 是重回  false 就是不重回:丢弃消息，如果重回队列的话，异常没有解决，就会进入死循环
                            //channel.basicNack(envelope.getDeliveryTag(),false,true);
                        } catch (Exception e2) {
                            e2.printStackTrace();
                        }
                    }
                }
            };
            //参数autoAck=false,启动手动ACK，此时消息接收后，但又不手动发送ACK确认，消息会一直停留在队列中，可能造成消息的重复获取
            channel.basicConsume(QUEUE_NAME_ACK, false, consumer);
        } catch (Exception e3) {
            e3.printStackTrace();
        }
    }


    //prefetchCount(缺省值250)：每个消费者一次性从队列中取出的消息个数,提高这个参数可以减少了网络传输的时间，对于一些严格要求顺序的消息或者
    //希望在多个使用者之间实现更均匀的消息分布的prefetch的值应当设置为1。通过channel.basicQos来设置
    //限流：prefetchCount的值就是消费者可以处理的最大消息数量（未确认），超过此数，队列会暂停发送消息给该消费者，直到消费者开始ack。
    //concurrentConsumers（缺省值：1）：同一个消费者的并发数（线程数），可以提高消息的消费能力,防止消息的堆积
    private static final String QUEUE_NAME_LIMIT = "queue_limit";
   //--------------------限流--------------------
    public static void limitFlowProducer() {
        try {
            Connection connection = BasicDemo.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME_LIMIT, false, false, false, null);
            for (int i = 0; i < count; i++) {
                channel.basicPublish("", QUEUE_NAME_LIMIT, null, ("消费端限流数据：" + (i + 1)).getBytes());
            }
            channel.close();
            connection.close();

        } catch (
                Exception e) {
            e.printStackTrace();
        }
    }

    public static void limitFlowConsumer() {
        try {
            Connection connection = BasicDemo.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME_LIMIT, false, false, false, null);
            /**  设置限流机制
             *  param1: prefetchSize，消息本身的大小 如果设置为0  那么表示对消息本身的大小不限制
             *  param2: prefetchCount，告诉rabbitmq不要一次性给消费者推送大于N个消息
             *  param3：global，是否将上面的设置应用于整个通道，false表示只应用于当前消费者
             */
            channel.basicQos(0, 3, false);
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("[限流消费者]received ===> " + new String(body));

                    //限流情况 ack 不能设置自动签收，一定要手动签收
                    //本ack语句注释后，可以在管理后台看到queue_limit中有3条消息Uacked
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            channel.basicConsume(QUEUE_NAME_LIMIT, false, defaultConsumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //交换机、队列、消息都要持久化,则可以保证消息在服务端不丢失，即使消费者在获取消息之前mq服务宕机
    //channel.exchangeDeclare("EXCHANGE_NAME_DIRECT", BuiltinExchangeType.DIRECT,true);  durable=true
    //channel.queueDeclare(QUEUE_NAME_DURABLE, true, false, false, null); durable=true
    //channel.basicPublish("", QUEUE_NAME_DURABLE, MessageProperties.PERSISTENT_TEXT_PLAIN,"msg");BasicProperties中的deliveryMode属性设置为2
    private static final String QUEUE_NAME_DURABLE = "queue_durable";
    //--------------------持久化--------------------
    public static void durableProducer() {
        try {
            Connection connection = BasicDemo.getConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare("EXCHANGE_DIRECT_DURABLE", BuiltinExchangeType.DIRECT,true);
            channel.queueDeclare(QUEUE_NAME_DURABLE, true, false, false, null);

            for (int i = 0; i < count; i++) {
                channel.basicPublish("EXCHANGE_DIRECT_DURABLE", QUEUE_NAME_DURABLE, MessageProperties.PERSISTENT_TEXT_PLAIN, ("持久化数据：" + (i + 1)).getBytes());
            }
            channel.close();
            connection.close();

        } catch (
                Exception e) {
            e.printStackTrace();
        }
    }

    public static void durableConsumer() {
        try {
            Connection connection = BasicDemo.getConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME_DURABLE, true, false, false, null);
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("[持久化消费者]received ===> " + new String(body));
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            channel.basicConsume(QUEUE_NAME_DURABLE, true, defaultConsumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //--------------------死信--------------------
    //1.生产者中在普通消息队列中发送消息，到达一定条件会产生死信，它们被自动转到绑定的死信交换器
    //2.普通消息队列消费者中，在队列定义时绑定死信交换机和死信routingKey
    //3.死信消息队列消费者中，定义死信交换器、队列，使用发送着的routingKey或绑定的死信routingKey将二者绑定
    public static void deadProducer() {
        String exchangeName = "normal_exchange";
        String routingKey = "dlx.normal";
        String msg = "Hello ，RabbitMQ 死信！";
        try {
        connection = BasicDemo.getConnection();
        channel = connection.createChannel();
        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .deliveryMode(2)
                    .contentEncoding("UTF-8")
                    .expiration("5000")
                    .build();

            //发送消息，下列情况会形成死信
            //1）消息被拒绝(basic.reject / basic.nack)，并且requeue = false
            //2）消息TTL过期，Time To Live已经过期
            //3）队列达到最大长度，queue的"x-max-length"被超越
            channel.basicPublish(exchangeName, routingKey, true, properties, msg.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void normalConsumer() {
        try {
            connection = BasicDemo.getConnection();
            channel = connection.createChannel();
            // 声明一个普通的交换机 和 队列 以及路由
            String exchangeName = "normal_exchange";
            String routingKey = "dlx.normal";
            String queueName = "normal_queue";
            channel.exchangeDeclare(exchangeName, "direct", true, false, null);

            //指定死信发送的Exchange
            Map<String, Object> agruments = new HashMap<String, Object>();
            agruments.put("x-dead-letter-exchange", "dead_exchange");         //设定死信时转发的交换器（死信交换器）
            agruments.put("x-dead-letter-routing-key", "dead_routing_key");   //死信转发到私信交换器时routingKey改为这里的设定值，否则使用原来的值
            //agruments.put("alternate-exchange", "backup_exchange");         //使用备份交换器可以部分替代死信交换器功能
            //正常队列的死信转到死信交换器上，这个agruments属性要设置到声明队列上
            channel.queueDeclare(queueName, true, false, false, agruments);
            channel.queueBind(queueName, exchangeName, routingKey);


            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    try {
                        int i=1/0;  //制造异常
//                        try {
//                            Thread.sleep(6000);
//                        } catch (InterruptedException e) {
//                            e.printStackTrace();
//                        }
                        System.out.println("[正常队列收]received ===> " + new String(body));
                        //手动正常签收
                        channel.basicAck(envelope.getDeliveryTag(), false);

                    } catch (Exception e) {
                    //拒绝消费消息且第三个参数false,表示不会重回队列,这样该消息将进入死信队列,
                    channel.basicNack(envelope.getDeliveryTag(), false, false);
                  }
                }
            };
            channel.basicConsume(queueName, false, defaultConsumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void deadConsumer() {
        try {
            connection = BasicDemo.getConnection();
            channel = connection.createChannel();
            // 声明一个普通的交换机 和 队列 以及路由
            String exchangeName = "dead_exchange";
            String routingKey = "dead_routing_key";
            String queueName = "dead_queue";

            //死信交换器的声明
            channel.exchangeDeclare(exchangeName,"direct",true, false, false, null);
            //死信队列的声明
            channel.queueDeclare(queueName, true, false, false, null);
            channel.queueBind(queueName, exchangeName, routingKey);

            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("[死信队列]received ===> " + new String(body));
//                    try {
//                        Thread.sleep(200);
//                    } catch (InterruptedException e) {
//                        e.printStackTrace();
//                    }
                }
            };
            channel.basicConsume(queueName, true, defaultConsumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //--------------------延迟消息队列--------------------
    //1.下载、安装和启动消息延迟插件；https://www.rabbitmq.com/community-plugins.html
    //2.发送消息时使用携带x-delay参数的header，其中定义延迟时间
    //3.接收消息时使用x-delayed-type的交换器
    public static void delayProducer() {
        String exchangeName = "delay_exchange";
        String routingKey = "delay_key";
        String msg = "Hello ，RabbitMQ 延迟消息！";
        try {
            connection = BasicDemo.getConnection();
            channel = connection.createChannel();

            //设置延迟消息参数
            Map<String, Object> delayHeaders = new HashMap<String, Object>();
            delayHeaders.put("x-delay", 5000);
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .headers(delayHeaders)
                    .contentEncoding("UTF-8")
                    .build();
            channel.basicPublish(exchangeName, routingKey, true, properties, msg.getBytes());
            System.out.println("延迟消息发送时间"+LocalDateTime.now());

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void delayConsumer() {
        try {
            connection = BasicDemo.getConnection();
            channel = connection.createChannel();
            String exchangeName = "delay_exchange";
            String routingKey = "delay_key";
            String msg = "Hello ，RabbitMQ 延迟消息！";
            String queueName = "delay_queue";
            //创建延迟交换器
            Map<String, Object> args = new HashMap<String, Object>();
            args.put("x-delayed-type", "direct");
            channel.exchangeDeclare(exchangeName, "x-delayed-message", true, false, args);
            channel.queueDeclare(queueName, true, false, false, null);
            channel.queueBind(queueName, exchangeName, routingKey);

            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    try {
                        System.out.println("延迟消息接收时间"+LocalDateTime.now());
                        System.out.println("[延迟队列]received ===> " + new String(body));
                        //手动正常签收
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    } catch (Exception e) {
                        //拒绝消费消息且第三个参数false,表示不会重回队列,这样该消息将进入死信队列,
                        channel.basicNack(envelope.getDeliveryTag(), false, false);
                    }
                }
            };
            channel.basicConsume(queueName, true, defaultConsumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
